﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;

namespace Cluster
{
    public abstract partial class ClusterManager
    {
        /// <summary>
        /// Node metadata
        /// </summary>
        [Serializable]
        protected class ClusterNode
        {
            [NonSerialized]
            public string ZkPath;       // name iz /cluster tree

            public Guid Id;             // the node unique id

            public string Name;         // server name

            public int ReplicasNumber;  // the number of replicated servers by this node
        }

        private PathWatcher clusterWatcher;                     // watches changes in the root

        protected List<ClusterNode> ClusterView { get; set; }   // the local view of the cluster

        protected ClusterNode SelfNode { get; set; }            // our node in the cluster

        public event EventHandler ClusterViewChanged;           // called on each local cluster view change (node added/removed)

        /// <summary>
        /// Updates the local view of the cluster from the Zoo Keeper server.
        /// On change, ClusterViewChanged event is called.
        /// </summary>
        [MethodImpl(MethodImplOptions.Synchronized)]
        private void UpdateClusterViewAndNotify()
        {
            if (UpdateClusterView())
            {
                Debug("Cluster view change detected");

                if (ClusterViewChanged != null)
                {
                    ClusterViewChanged(this, EventArgs.Empty);
                }
            }
        }

        /// <summary>
        /// Updates the local view of the cluster from the Zoo Keeper server
        /// </summary>
        /// <returns></returns>
        private bool UpdateClusterView()
        {
            Debug("Updating cluster view");

            var view = new List<ClusterNode>();

            while (true) // the nodes might dissappear while retrieving their data so we need to retry
            {
                view.Clear();

                var nodes = GetZkChildren(ZkClusterPath);

                try
                {
                    // get meta-data for each node
                    foreach (var node in nodes)
                    {
                        var data = GetZkNodeData(ZkClusterPath + "/" + node);
                        if (data != null)
                        {
                            var cn = Deserialize<ClusterNode>(data);
                            cn.ZkPath = node;

                            view.Add(cn);

                            if (SelfNode != null && cn.Id == SelfNode.Id)
                            {
                                SelfNode = cn;
                            }
                        }
                    }

                    break;
                }
                catch (ClusterException)
                {
                    Debug("Cluster view was changed during node data retrival, retrying");
                }
            }

            // sort by name so we can compare to the previous view and detect changes
            view.Sort((cn1, cn2) => String.Compare(cn1.ZkPath, cn2.ZkPath, StringComparison.Ordinal));

            // check if the current view changed
            bool changed = view.Count != ClusterView.Count ||
                           view.Where((t, i) => t.ZkPath != ClusterView[i].ZkPath).Any();

            Debug("Retrieved cluster view ({0} nodes)", view.Count);

            // print the retrieved view to log - but only if leader (to avoid huge logs)
            if (isLeader)
            {
                var sb = new StringBuilder();
                sb.AppendLine("Cluster view:");

                foreach (var node in view)
                {
                    sb.AppendLine(string.Format("\t{0}\t name={1}\t replicas={2}",
                        node.ZkPath, node.Name, node.ReplicasNumber));
                }

                Debug(sb.ToString());
            }

            ClusterView = view;

            return changed;
        }

        /// <summary>
        /// Registers our node to the cluster
        /// </summary>
        [MethodImpl(MethodImplOptions.Synchronized)]
        private void RegisterNode()
        {
            // set watch on the cluster path
            WatchClusterPath();

            // our meta data
            SelfNode = CreateClsuterNode();
            SelfNode.Name = ServerName;
            SelfNode.ReplicasNumber = 1;
            SelfNode.Id = uniqueId;

            // create the Zoo Keeper node
            var zpath = CreateZkNode(ZkClusterPath + "/", false, true, Serialize(SelfNode));

            SelfNode.ZkPath = zpath.Substring(ZkClusterPath.Length + 1);
            
            Debug("Registered self to cluster as " + zpath);
        }

        /// <summary>
        /// Updates the meta-data of our node
        /// </summary>
        private void UpdateSelfNodeInCluster()
        {
            SetZkNodeData(ZkClusterPath + "/" + SelfNode.ZkPath, Serialize(SelfNode), -1);
        }

        /// <summary>
        /// Tracks changes in the cluster Zoo Keeper path
        /// </summary>
        private void WatchClusterPath()
        {
            if (clusterWatcher == null)
            {
                clusterWatcher = new PathWatcher(UpdateClusterViewAndNotify, WatchClusterPath);
            }

            WatchZkPath(ZkClusterPath, clusterWatcher);
        }

        /// <summary>
        /// Creates the cluster node structure.
        /// Can be overridden in derived classes to add more fields. 
        /// </summary>
        /// <returns></returns>
        protected virtual ClusterNode CreateClsuterNode()
        {
            return new ClusterNode();
        }

        /// <summary>
        /// The Zoo Keeper cluster path
        /// </summary>
        private string ZkClusterPath
        {
            get { return "/" + ClusterName + "/cluster"; }
        }
    }
}
